package com.shujia;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.filter.RowFilter;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import javax.swing.*;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.List;

public class Demo4Filter {
    Connection conn = null;
    Admin admin = null;
    Table students = null;
    TableName name = TableName.valueOf("students");

    public void useFilterAndPrint(Filter filter) throws IOException {
        Scan scan = new Scan();
        scan.setFilter(filter);
        for (Result rs : students.getScanner(scan)) {
            // result表示一条数据
            String rowkey = Bytes.toString(rs.getRow());
            System.out.print(rowkey + " ");
            List<Cell> cells = rs.listCells();
            for (Cell cell : cells) {
                String value = Bytes.toString(CellUtil.cloneValue(cell));
                System.out.print(value + " ");
            }
            System.out.println();
        }

    }

    @Before
    public void init() {
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "master:2181,node1:2181,node2:2181");
        try {
            conn = ConnectionFactory.createConnection(conf);
            admin = conn.getAdmin();
            students = conn.getTable(name);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    @Test
    // 通过RowFilter过滤比rowKey 1500100010 小的所有值出来
    public void BinaryComparatorFilter() throws IOException {
        BinaryComparator binaryComparator = new BinaryComparator("1500100010".getBytes());
        RowFilter rowFilter = new RowFilter(CompareFilter.CompareOp.LESS, binaryComparator);
        useFilterAndPrint(rowFilter);
    }

    @Test
    // 通过FamilyFilter查询列簇名包含in的所有列簇下面的数据
    public void SubstringComparatorFilter() throws IOException {
        SubstringComparator comparator = new SubstringComparator("in");
        FamilyFilter familyFilter = new FamilyFilter(CompareFilter.CompareOp.EQUAL, comparator);
        useFilterAndPrint(familyFilter);
    }

    // 查询列名包含a的列 下面的所有的值
    // QualifierFilter SubstringComparator
    @Test
    public void SubstringComparatorQualifierFilter() throws IOException {
        SubstringComparator comparator = new SubstringComparator("a");
        QualifierFilter qualifierFilter = new QualifierFilter(CompareFilter.CompareOp.EQUAL, comparator);
        useFilterAndPrint(qualifierFilter);
    }

    // 过滤文科一班的学生
    // ValueFilter BinaryComparator
    // ValueFilter是作用在每一个cell上，只有符合条件的cell才会保留
    @Test
    public void BinaryComparatorValueFilter() throws IOException {
        BinaryComparator binaryComparator = new BinaryComparator("文科一班".getBytes());
        ValueFilter valueFilter = new ValueFilter(CompareFilter.CompareOp.EQUAL, binaryComparator);
        useFilterAndPrint(valueFilter);
    }

    // 过滤文科一班的学生 并且返回学生的所有的信息
    // 会完整地返回整条数据
    // 在比较地时候需要指定 列簇和列名，如果数据中存在没有所指定的列簇的数据 则会保留并返回
    // 在比较地时候还需要指定 列簇和列名，如果数据中存在没有所指定的列名的数据 则也会保留并返回
    @Test
    public void SingleColumnValueFilterClazz() throws IOException {
        SingleColumnValueFilter singleColumnValueFilter = new SingleColumnValueFilter(
                "info".getBytes()
                , "clazz".getBytes()
                , CompareFilter.CompareOp.EQUAL
                , "文科一班".getBytes());
        useFilterAndPrint(singleColumnValueFilter);
    }

    // 相对于SingleColumnValueFilter 会将用于过滤的列值去除 并返回
    @Test
    public void SingleColumnValueExcludeFilterClazz() throws IOException {
        SingleColumnValueExcludeFilter singleColumnValueExcludeFilter = new SingleColumnValueExcludeFilter(
                "info".getBytes()
                , "clazz".getBytes()
                , CompareFilter.CompareOp.EQUAL
                , "文科一班".getBytes());
        useFilterAndPrint(singleColumnValueExcludeFilter);
    }

    @Test
    // 通过PrefixFilter查询以150010008开头的所有前缀的rowkey
    // 相当于 BinaryPrefixComparator加上RowFilter
    public void PrefixFilterFilter() throws IOException {
        PrefixFilter prefixFilter = new PrefixFilter("150010008".getBytes());
        useFilterAndPrint(prefixFilter);
    }

    @Test
    // 通过PageFilter查询第三页的数据，每页10条
    public void PageFilter() throws IOException {
        int page = 3;
        int pageSize = 10;

        // 计算出当前页面的第一条数据的位置（相对于整个数据的第一条）
        // 例如 想取第三页的数据 那么第三页第一条数据的位置 应该是 21
        int current_page_first_size = (page - 1) * pageSize + 1;
        PageFilter pageFilter = new PageFilter(current_page_first_size);
        Scan scan = new Scan();
        scan.setFilter(pageFilter);
        byte[] rk = null;
        for (Result result : students.getScanner(scan)) {
            // 取出当前页面的第一条数据的rowkey
            rk = result.getRow();
        }

        PageFilter pageFilter1 = new PageFilter(pageSize);
        Scan scan1 = new Scan();
        scan1.withStartRow(rk);
        scan1.setFilter(pageFilter1);
        for (Result rs : students.getScanner(scan1)) {
            // result表示一条数据
            String rowkey = Bytes.toString(rs.getRow());
            System.out.print(rowkey + " ");
            List<Cell> cells = rs.listCells();
            for (Cell cell : cells) {
                String value = Bytes.toString(CellUtil.cloneValue(cell));
                System.out.print(value + " ");
            }
            System.out.println();
        }


    }

    @Test
    // 通过合理的设计rowkey查询第五页的数据，每页10条
    // students表中的rowkey是从1500100001依次递增至1500101000
    // 所以第五页数据的rowkey范围可以直接计算出来
    // 1500100041 ~ 1500100050
    public void PageFilterWithRowkey() throws IOException {
        String rowkeyPrefix = "1500100000";
        int page = 5;
        int pageSize = 10;

        int current_page_start_row_offset = (page - 1) * pageSize + 1;
        int i = Integer.parseInt(rowkeyPrefix);
        int start_rowkey = i + current_page_start_row_offset;
        int end_rowkey = start_rowkey + pageSize;
        String start_rowkey_str = start_rowkey + "";
        String end_rowkey_str = end_rowkey + "";

        Scan scan = new Scan();
        scan.withStartRow(start_rowkey_str.getBytes());
        scan.withStopRow(end_rowkey_str.getBytes());
        for (Result rs : students.getScanner(scan)) {
            // result表示一条数据
            String rowkey = Bytes.toString(rs.getRow());
            System.out.print(rowkey + " ");
            List<Cell> cells = rs.listCells();
            for (Cell cell : cells) {
                String value = Bytes.toString(CellUtil.cloneValue(cell));
                System.out.print(value + " ");
            }
            System.out.println();
        }


    }

    @Test
    // 查询文科班中的学生中学号以150010008开头 并且 年龄小于23的学生信息
    public void FilterListFilter() throws IOException {
        PrefixFilter prefixFilter = new PrefixFilter("150010008".getBytes());
        SingleColumnValueFilter singleColumnValueFilter = new SingleColumnValueFilter("info".getBytes(), "age".getBytes(), CompareFilter.CompareOp.LESS, "23".getBytes());
        SingleColumnValueFilter singleColumnValueFilter2 = new SingleColumnValueFilter("info".getBytes(), "clazz".getBytes(), CompareFilter.CompareOp.EQUAL, "文科一班".getBytes());

        // 通过FilterList将多个过滤器组装起来
        // 实际上FilterList也是Filter的子类
        FilterList filterList = new FilterList();
        filterList.addFilter(prefixFilter);
        filterList.addFilter(singleColumnValueFilter);
        filterList.addFilter(singleColumnValueFilter2);
        useFilterAndPrint(filterList);

    }

    @After
    public void close() {
        try {
            admin.close();
            conn.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

}
